package com.cmdfut.rxjava.rxjava;

import android.os.Handler;
import android.os.Looper;

public class ObservableObserveOn<T> extends AbstractObservableWithUpstream<T, T> {
    public ObservableObserveOn(ObservableSource<T> source) {
        super(source);
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        ObserverOnObserver<T> parent = new ObserverOnObserver<>(observer);
        source.subscribe(parent);
    }

    //包装下游观察者，并关联onNext,.....  放入主线程中执行
    static final class ObserverOnObserver<T> implements Observer<T>, Disposable {
        final Observer<? super T> actual;
        Disposable disposeble;
        private Handler handler;

        ObserverOnObserver(Observer<? super T> actual) {
            this.actual = actual;
            handler = new Handler(Looper.getMainLooper());
        }

        @Override
        public void dispose(boolean bool) {
            this.disposeble.dispose(bool);
        }

        @Override
        public boolean isDisposed() {
            return this.disposeble.isDisposed();
        }

        @Override
        public void onSubscribe(Disposable d) {
            this.disposeble = d;
            actual.onSubscribe(d);
        }

        @Override
        public void onNext(final T t) {
            handler.post(new Runnable() {
                @Override
                public void run() {
                    actual.onNext(t);
                }
            });
        }

        @Override
        public void onError(final Throwable e) {
            handler.post(new Runnable() {
                @Override
                public void run() {
                    actual.onError(e);
                }
            });
        }

        @Override
        public void onComplete() {
            handler.post(new Runnable() {
                @Override
                public void run() {
                    actual.onComplete();
                }
            });
        }
    }
}

